package com.lifeverse.service;

import com.lifeverse.event.LifeEntityEvent;
import com.lifeverse.util.JsonUtils;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;

import java.util.concurrent.CompletableFuture;

/**
 * 事件发布服务
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class EventPublisher {
    
    private final KafkaTemplate<String, String> kafkaTemplate;
    
    // Kafka主题常量
    public static final String LIFE_ENTITY_INTERACTION_TOPIC = "life-entity-interaction";
    public static final String NETWORK_HEALTH_TOPIC = "network-health";
    public static final String GROUP_DECISION_TOPIC = "group-decision";
    public static final String CONSCIOUSNESS_STATE_TOPIC = "consciousness-state";
    public static final String SYSTEM_ALERT_TOPIC = "system-alert";
    
    /**
     * 发布生命体事件
     */
    public CompletableFuture<SendResult<String, String>> publishEvent(LifeEntityEvent event) {
        return publishEvent(getTopicForEvent(event), event);
    }
    
    /**
     * 发布事件到指定主题
     */
    public CompletableFuture<SendResult<String, String>> publishEvent(String topic, LifeEntityEvent event) {
        try {
            String eventJson = JsonUtils.toJson(event);
            String key = generateEventKey(event);
            
            log.debug("发布事件到主题 {}: eventId={}, eventType={}, sourceEntityId={}", 
                    topic, event.getEventId(), event.getEventType(), event.getSourceEntityId());
            
            CompletableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, key, eventJson);
            
            future.whenComplete((result, ex) -> {
                if (ex == null) {
                    log.debug("事件发布成功: eventId={}, topic={}, partition={}, offset={}", 
                            event.getEventId(), topic, 
                            result.getRecordMetadata().partition(),
                            result.getRecordMetadata().offset());
                } else {
                    log.error("事件发布失败: eventId={}, topic={}", event.getEventId(), topic, ex);
                }
            });
            
            return future;
        } catch (Exception e) {
            log.error("发布事件时发生异常: eventId={}, topic={}", event.getEventId(), topic, e);
            CompletableFuture<SendResult<String, String>> failedFuture = new CompletableFuture<>();
            failedFuture.completeExceptionally(e);
            return failedFuture;
        }
    }
    
    /**
     * 批量发布事件
     */
    public void publishEvents(String topic, LifeEntityEvent... events) {
        for (LifeEntityEvent event : events) {
            publishEvent(topic, event);
        }
    }
    
    /**
     * 发布高优先级事件
     */
    public CompletableFuture<SendResult<String, String>> publishHighPriorityEvent(LifeEntityEvent event) {
        event.setPriority(LifeEntityEvent.EventPriority.HIGH);
        return publishEvent(event);
    }
    
    /**
     * 发布紧急事件
     */
    public CompletableFuture<SendResult<String, String>> publishUrgentEvent(LifeEntityEvent event) {
        event.setPriority(LifeEntityEvent.EventPriority.URGENT);
        return publishEvent(event);
    }
    
    /**
     * 发布关键事件
     */
    public CompletableFuture<SendResult<String, String>> publishCriticalEvent(LifeEntityEvent event) {
        event.setPriority(LifeEntityEvent.EventPriority.CRITICAL);
        return publishEvent(event);
    }
    
    /**
     * 根据事件类型获取对应的主题
     */
    private String getTopicForEvent(LifeEntityEvent event) {
        switch (event.getEventType()) {
            case "LIFE_ENTITY_INTERACTION":
                return LIFE_ENTITY_INTERACTION_TOPIC;
            case "NETWORK_HEALTH":
                return NETWORK_HEALTH_TOPIC;
            case "GROUP_DECISION":
                return GROUP_DECISION_TOPIC;
            case "CONSCIOUSNESS_STATE_CHANGE":
                return CONSCIOUSNESS_STATE_TOPIC;
            case "SYSTEM_ALERT":
                return SYSTEM_ALERT_TOPIC;
            default:
                log.warn("未知事件类型: {}, 使用默认主题", event.getEventType());
                return "default-topic";
        }
    }
    
    /**
     * 生成事件键
     */
    private String generateEventKey(LifeEntityEvent event) {
        if (event.getSourceEntityId() != null) {
            return event.getEventType() + "_" + event.getSourceEntityId();
        }
        return event.getEventType() + "_" + event.getEventId();
    }
    
    /**
     * 发布系统通知
     */
    public void publishSystemNotification(String message, LifeEntityEvent.EventPriority priority) {
        LifeEntityEvent notification = new LifeEntityEvent("SYSTEM_NOTIFICATION", null) {};
        notification.setPriority(priority);
        notification.getEventData().put("message", message);
        notification.getEventData().put("timestamp", System.currentTimeMillis());
        
        publishEvent(SYSTEM_ALERT_TOPIC, notification);
    }
    
    /**
     * 发布错误事件
     */
    public void publishErrorEvent(String errorMessage, String errorCode, Long entityId) {
        LifeEntityEvent errorEvent = new LifeEntityEvent("SYSTEM_ERROR", entityId) {};
        errorEvent.setPriority(LifeEntityEvent.EventPriority.HIGH);
        errorEvent.getEventData().put("errorMessage", errorMessage);
        errorEvent.getEventData().put("errorCode", errorCode);
        errorEvent.getEventData().put("timestamp", System.currentTimeMillis());
        
        publishEvent(SYSTEM_ALERT_TOPIC, errorEvent);
    }
    
    /**
     * 发布警告事件
     */
    public void publishWarningEvent(String warningMessage, Long entityId) {
        LifeEntityEvent warningEvent = new LifeEntityEvent("SYSTEM_WARNING", entityId) {};
        warningEvent.setPriority(LifeEntityEvent.EventPriority.NORMAL);
        warningEvent.getEventData().put("warningMessage", warningMessage);
        warningEvent.getEventData().put("timestamp", System.currentTimeMillis());
        
        publishEvent(SYSTEM_ALERT_TOPIC, warningEvent);
    }
    
    /**
     * 检查Kafka连接状态
     */
    public boolean isKafkaAvailable() {
        try {
            // 尝试发送一个测试消息
            LifeEntityEvent testEvent = new LifeEntityEvent("HEALTH_CHECK", null) {};
            testEvent.getEventData().put("test", true);
            
            CompletableFuture<SendResult<String, String>> future = publishEvent("health-check", testEvent);
            future.get(java.util.concurrent.TimeUnit.SECONDS.toMillis(5), java.util.concurrent.TimeUnit.MILLISECONDS);
            return true;
        } catch (Exception e) {
            log.warn("Kafka连接检查失败", e);
            return false;
        }
    }
}